package com.niodata.dp.monitor.components.task.controller;

import com.google.common.base.Strings;
import com.niodata.dp.monitor.common.util.ThrowableUtil;
import com.niodata.dp.monitor.components.task.service.TaskAlertService;
import com.niodata.dp.monitor.core.task.TaskExecutionManager;
import com.niodata.dp.monitor.core.task.TaskManagerImpl;
import com.niodata.dp.monitor.sharelib.entity.az.AzkabanSummary;
import com.niodata.dp.monitor.sharelib.entity.task.TaskExecResult;
import com.niodata.dp.monitor.sharelib.entity.task.TaskMonitorResult;
import com.niodata.dp.util.StringUtil;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Date;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.Collectors;
import org.apache.commons.lang3.StringUtils;
import org.apache.ibatis.annotations.Param;
import org.apache.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.format.annotation.DateTimeFormat;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;


/**
 * Created by ndp on 17-7-10.
 */
@RestController
@RequestMapping("/monitor/task")
public class TaskAlertController {

  @Autowired
  private TaskAlertService taskMonitorService;

  private Logger logger = Logger.getLogger(TaskAlertController.class);

  @Autowired
  @Qualifier("taskManagerImpl")
  private TaskExecutionManager taskManager;

  private ExecutorService executorService = Executors.newSingleThreadScheduledExecutor();

  @RequestMapping(value = "/alerts", method = RequestMethod.GET)
  public TaskMonitorResult taskAlertFilter(
        @RequestParam(value = "userId", required = false, defaultValue = "1") long userId,
        @RequestParam(value = "groupIds", required = false, defaultValue = "") String groupIds,
        @RequestParam(value = "pageIndex", required = false, defaultValue = "1") int pageIndex,
        @RequestParam(value = "pageSize", required = false, defaultValue = "50") int pageSize,
        @RequestParam(value = "taskId", required = false, defaultValue = "") String taskId,
        @RequestParam(value = "status", required = false, defaultValue = "") String status) {

    TaskMonitorResult monitorResult;
    try {
      monitorResult = taskMonitorService
            .taskAlertMsgsFilter(userId, groupIds, pageIndex, pageSize, taskId, status);
    } catch (Throwable throwable) {
      monitorResult = new TaskMonitorResult();
      monitorResult.setSuccess(false);
      monitorResult.setErrorMsg(throwable.getMessage());
      monitorResult.setErrorDetail(ThrowableUtil.getErrorDetailFromThrowable(throwable));
      logger.error("get task alerts", throwable);
    }
    return monitorResult;
  }

  @RequestMapping(value = "/recentfinish", method = RequestMethod.GET)
  public TaskExecResult recentlyFinishedExecs(
        @RequestParam(value = "userId", required = false, defaultValue = "1") long userId,
        @RequestParam(value = "groupIds", required = false, defaultValue = "") String groupIds) {

    List<Long> ids = new ArrayList<>();
    if (!Strings.isNullOrEmpty(groupIds.trim())) {
      ids = Arrays.stream(groupIds.split(",")).map(str -> {
        return Long.valueOf(str);
      }).collect(Collectors.toList());
    }
    return taskMonitorService.getRecentlyFinishedExec(userId, ids);
  }

  /**
   * get execs.
   *
   * @param userId userid
   * @param groupIds group ids
   * @param pageIndex page index
   * @param pageSize page size
   * @param status status
   * @return exec result
   */
  @RequestMapping(value = "/execs", method = RequestMethod.GET)
  public TaskExecResult taskExecFilter(
        @RequestParam(value = "taskId", required = false) String taskId,
        @RequestParam(value = "userId", required = false, defaultValue = "1") long userId,
        @RequestParam(value = "groupIds", required = false, defaultValue = "") String groupIds,
        @RequestParam(value = "pageIndex", required = false, defaultValue = "1") int pageIndex,
        @RequestParam(value = "pageSize", required = false, defaultValue = "50") int pageSize,
        @RequestParam(value = "status", required = false, defaultValue = "") String status,
        @RequestParam(value = "startTime", required = false) String startTime,
        @RequestParam(value = "endTime", required = false) String endTime) {

    TaskExecResult taskExecResult;

    if ((StringUtils.isBlank(startTime) && !StringUtils.isBlank(endTime))
          || (!StringUtils.isBlank(startTime) && StringUtils.isBlank(endTime))) {
      taskExecResult = new TaskExecResult();
      taskExecResult.setSuccess(false);
      taskExecResult.setErrorMsg("param is error!");
      return taskExecResult;
    }

    if (StringUtil.isEmpty(taskId)) {
      taskId = null;
    }
    try {
      List<Long> ids = new ArrayList<>();
      if (!Strings.isNullOrEmpty(groupIds.trim())) {
        ids = Arrays.stream(groupIds.split(",")).map(str -> {
          return Long.valueOf(str);
        }).collect(Collectors.toList());
      }
      taskExecResult = taskMonitorService
            .getTaskExecResult(taskId, userId, ids, pageIndex, pageSize, status, startTime,
                  endTime);
    } catch (Throwable throwable) {
      taskExecResult = new TaskExecResult();
      taskExecResult.setSuccess(false);
      taskExecResult.setErrorMsg(throwable.getMessage());
      taskExecResult.setErrorDetail(ThrowableUtil.getErrorDetailFromThrowable(throwable));
      logger.error("get task alerts", throwable);
    }
    return taskExecResult;
  }

  /**
   * getAzKaBanStatus.
   */
  @RequestMapping(value = "/azkaban/status", method = RequestMethod.GET)
  public AzkabanSummary getAzKaBanStatus(@Param(value = "startTime") @DateTimeFormat(
        pattern = "yyyy-MM-dd HH:mm:ss") Date startTime, @Param(value = "stopTime") @DateTimeFormat(
        pattern = "yyyy-MM-dd HH:mm:ss") Date stopTime) {
    AzkabanSummary azkabanSummary = taskMonitorService.getAzkabanSummary(startTime, stopTime);
    return azkabanSummary;
  }


  @RequestMapping(value = "/syncExecutions", method = RequestMethod.POST)
  public String syncExecutionList() {
    logger.info("sync executions....");
    Runnable runnable = new Runnable() {
      @Override
      public void run() {
        taskManager.syncExecutionList();
      }
    };
    executorService.submit(runnable);
    return "ok";
  }
}
